package com.swak.ringbuffer;

import com.swak.utils.ring.RingBuffer;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;

/**
 * 
 * 数据分片
 * 
 * @author 618lf
 */
@Getter
@Setter
@Accessors(chain = true)
@ToString(exclude = { "data" })
class Segment {
	private long sn;
	private byte acks;
	private Object data;
}

/**
 * 通过写序号和读序号来控制数据;
 * 
 * @author 618lf
 */
public class Window extends RingBuffer<Segment> {

	private long sendUna = 0;
	private long sendNxt = 0;
	private final int size;
	private int wnd;

	public Window(int size, int wnd) {
		super(size);
		this.size = this.bufferSize;
		this.wnd = wnd;
	}

	boolean isFull() {
		int write = (int) (sendNxt & indexMask);
		int read = (int) (sendUna & indexMask);
		return write == read && sendNxt > sendUna;
	}

	boolean isEmpty() {
		return sendNxt == sendUna;
	}

	public boolean send(Segment data) {
		if (this.isFull()) {
			return false;
		}
		if (this.sendNxt - this.sendUna >= wnd) {
			return false;
		}
		data.setSn(sendNxt++);
		this.put(data.getSn(), data);
		return true;
	}

	public boolean ack(long ackSn) {

		/*
		 * check ack sn
		 */
		if (!(ackSn >= sendUna && ackSn < sendNxt)) {
			return false;
		}

		/*
		 * update ack state
		 */
		Segment ackSegment = this.get(ackSn);
		if (ackSegment != null) {
			ackSegment.setAcks((byte) (ackSegment.getAcks() + 1));
		}

		/*
		 * move acks;
		 */
		for (; sendUna < sendNxt;) {
			Segment segment = this.get(sendUna);
			if (segment.getAcks() >= 1) {
				sendUna++;
				continue;
			}
			break;
		}
		return true;
	}

	public void ack(long... ackSns) {

		/*
		 * batch ack
		 */
		for (long ackSn : ackSns) {
			if (ackSn >= sendUna && ackSn < sendNxt) {
				Segment ackSegment = this.get(ackSn);
				if (ackSegment != null) {
					ackSegment.setAcks((byte) (ackSegment.getAcks() + 1));
				}
			}
		}

		/*
		 * move acks;
		 */
		for (; sendUna < sendNxt;) {
			Segment segment = this.get(sendUna);
			if (segment.getAcks() >= 1) {
				sendUna++;
				continue;
			}
			break;
		}
	}

	@Override
	public String toString() {
		StringBuilder buffer = new StringBuilder();
		buffer.append("[").append("sendUna=").append(this.sendUna).append(", sendNxt=").append(this.sendNxt)
				.append(", size=").append(this.size).append("]\n");
		Object[] entries = this.entries();
		int i = 0;
		for (Object entrie : entries) {
			if (entrie != null) {
				buffer.append(i++).append("：").append(entrie.toString()).append("\n");
			}
		}
		return buffer.toString();
	}

	public static void main(String[] args) {

		/*
		 * 创建一个窗口大小 8 ； 可发送窗口 10（其实最多也只能发送8个数据）
		 */
		Window window = new Window(2, 2);

		/*
		 * 写满数据
		 */
		for (;;) {
			boolean flag = window.send(new Segment());
			if (!flag) {
				break;
			}
		}
		System.out.println(window);

		/*
		 * 确认数据： 不一定是现有的数据
		 */
		System.out.println("#######  确认 #######");
		window.ack(2);
		window.ack(10);
		System.out.println(window);

		/*
		 * 确认数据
		 */
		window.ack(0);
		System.out.println(window);

		/*
		 * 写满数据
		 */
		System.out.println("################ 再次写满数据 ############");
		for (;;) {
			boolean flag = window.send(new Segment());
			if (!flag) {
				break;
			}
		}
		System.out.println(window);
		System.out.println("################ 全部ACK ############");
		window.ack(1, 2, 3, 4, 5, 6);
		System.out.println(window);

		/*
		 * 写满数据
		 */
		for (;;) {
			boolean flag = window.send(new Segment());
			if (!flag) {
				break;
			}
		}
		System.out.println(window);
	}
}
